#include "access/pushpage.h"

#include "postgres.h"
#include "utils/elog.h"
#include "utils/guc.h"
#include "utils/builtins.h"
#include "libpq-fe.h"
#include "lib/stringinfo.h"
#include "access/xlog.h"
#include "utils/pg_lsn.h"


clock_t start_time;
XLogRecPtr	PushPtr = 0;
XLogRecPtr  ApplyLsn = 0;
XLogRecPtr PrePushPtr = 0;
XLogRecPtr CheckPointPtr = InvalidXLogRecPtr;
XLogRecPtr FileCheckPointPtr = InvalidXLogRecPtr;

//this for cut logindex
XLogRecPtr PrevPushPoint = InvalidXLogRecPtr;
XLogRecPtr LastPushPoint = InvalidXLogRecPtr;

CheckPoint GlobalCheckPoint;
uint8 GlobalState;


#define maxconnlen  1024
static PGconn *pushconn = NULL;
static PGconn *connToPushStandby = NULL;
pid_t startupPid = 0;


// static bool ConnectPushStandbyDB() {
// 	char	   *err;
// 	const char *keys[] = {"dbname","user","password","host","port",NULL};
// 	const char *vals[] = {"postgres","repl","123456","100.73.36.123","15431",NULL};
// 	connToPushStandby = PQconnectdbParams(keys, vals, false);
// 	if (PQstatus(connToPushStandby) == CONNECTION_BAD)
// 	{
// 		err = pchomp(PQerrorMessage(connToPushStandby));
// 		ereport(ERROR,
// 					(errcode(ERRCODE_CONNECTION_FAILURE),
// 					 errmsg("push standby could not connect to the push standby server: %s", err)));
// 		return false;
// 	}
// 	return true;
	
// }
static bool ConnectPrimaryDB(void);

static bool ConnectPrimaryDB(void) {
	char	   *err;
	char		conninfo[maxconnlen];
	// const char *keys[] = {"dbname","user","password","host","port",NULL};
	// const char *vals[] = {"postgres","repl","123456","100.73.36.123","15432",NULL};
	strlcpy(conninfo, (char *) PrimaryConnInfo, maxconnlen);
	/* Establish the connection to the primary for query Min Lsn*/
	/*
	 * We use the expand_dbname parameter to process the connection string (or
	 * URI), and pass some extra options.
	 */
	/* Note we do not want libpq to re-expand the dbname parameter */
	pushconn = PQconnectdb(conninfo);
	// pushconn = PQconnectdbParams(keys, vals, true);
	if (PQstatus(pushconn) == CONNECTION_BAD)
	{
		err = pchomp(PQerrorMessage(pushconn));
		ereport(WARNING,
					(errcode(ERRCODE_CONNECTION_FAILURE),
					 errmsg("push standby could not connect to the primary server: %s", err)));
		return false;
	}
	return true;	
}

bool ReConnectPrimaryDB(void) {
	if (push_standby == true && pushconn!=NULL) {
		PQfinish(pushconn);
		pushconn = NULL;
		if (ConnectPrimaryDB() == true) {
			return true;
		}
	}
	return false;
}


// static bool ConnectPrimaryDB4ReplyLSN() {
// 	char	   *err;
// 	char		conninfo[maxconnlen];
// 	const char *keys[] = {"dbname","user","password","host","port",NULL};
// 	const char *vals[] = {"postgres","postgres","","100.73.36.123","15432",NULL};
// 	strlcpy(conninfo, (char *) PrimaryConnInfo, maxconnlen);
// 	/* Establish the connection to the primary for query Min Lsn*/
// 	/*
// 	 * We use the expand_dbname parameter to process the connection string (or
// 	 * URI), and pass some extra options.
// 	 */
// 	/* Note we do not want libpq to re-expand the dbname parameter */
// 	pushconn = PQconnectdbParams(keys, vals, true);
// 	if (PQstatus(pushconn) == CONNECTION_BAD)
// 	{
// 		err = pchomp(PQerrorMessage(pushconn));
// 		ereport(WARNING,
// 					(errcode(ERRCODE_CONNECTION_FAILURE),
// 					 errmsg("push standby could not connect to the primary server: %s", err)));
// 		return false;
// 	}
// 	return true;	
// }


// XLogRecPtr QueryPushLsn() 
// {
// 	StringInfoData cmd;
// 	XLogRecPtr 		replylsn = InvalidXLogRecPtr;
// 	char       *replyptr;
// 	initStringInfo(&cmd);
// 	appendStringInfoString(&cmd,"select pg_last_wal_replay_lsn()");
// 	replylsn = InvalidXLogRecPtr;
// 	if (connToPushStandby == NULL) {
// 		if (ConnectPushStandbyDB() == false) {
// 			return InvalidXLogRecPtr;
// 		} 
// 	} 
// 	PGresult   *pgres = NULL;
// 	pgres = PQexec(connToPushStandby, cmd.data);
// 	if (PQresultStatus(pgres) == PGRES_TUPLES_OK && PQntuples(pgres) == 1) {
// 		replyptr = PQgetvalue(pgres, 0, 0);
// 		bool flag;
// 		replylsn = pg_lsn_in_internal(replyptr,&flag); 
		
// 	}
// 	PQfinish(connToPushStandby);
// 	connToPushStandby = NULL;
// 	PQclear(pgres);
// 	return replylsn;	
	
// }

XLogRecPtr QueryPushChkpointLsn(void) 
{
	ControlFileData *ControlFile;
	int			fd;
	int			r;
	XLogRecPtr	checkPoint;
	
	ControlFile = palloc(sizeof(ControlFileData));

	fd = BasicOpenFile(XLOG_CONTROL_FILE,
					   O_RDONLY | PG_BINARY);
	if (fd < 0)
		ereport(PANIC,
				(errcode_for_file_access(),
				 errmsg("could not open file \"%s\": %m",
						XLOG_CONTROL_FILE)));

	r = read(fd, ControlFile, sizeof(ControlFileData));
	if (r != sizeof(ControlFileData))
	{
		if (r < 0)
			ereport(PANIC,
					(errcode_for_file_access(),
					 errmsg("could not read file \"%s\": %m",
							XLOG_CONTROL_FILE)));
		else
			ereport(PANIC,
					(errcode(ERRCODE_DATA_CORRUPTED),
					 errmsg("could not read file \"%s\": read %d of %zu",
							XLOG_CONTROL_FILE, r, sizeof(ControlFileData))));
	}
	close(fd);
	checkPoint = ControlFile->checkPoint;
	pfree(ControlFile);
	
	return checkPoint;
}

XLogRecPtr QueryMinLsn(XLogRecPtr lsn) 
{
	StringInfoData cmd;
	XLogRecPtr 		replylsn;
	PGresult   *pgres = NULL;
	char       *replyptr;
	replylsn = InvalidXLogRecPtr;
	if (pushconn == NULL) {
		if (ConnectPrimaryDB() == false) {
			return InvalidXLogRecPtr;
		} 
	} 

	initStringInfo(&cmd);
	appendStringInfoString(&cmd, "SELECT t.application_name, t.replay_lsn, t.state, t.sync_state FROM pg_catalog.pg_stat_replication t WHERE t.application_name not like \'");
	appendStringInfoString(&cmd, "push%");
	appendStringInfoString(&cmd, "\' and t.application_name not like \'priv%\' order by t.replay_lsn limit 1");

	pgres = PQexec(pushconn, cmd.data);
	if (PQresultStatus(pgres) == PGRES_TUPLES_OK) {
		if (PQntuples(pgres) == 1) {
			replyptr = PQgetvalue(pgres, 0, 1);
			bool flag;
			replylsn = pg_lsn_in_internal(replyptr,&flag); 
			if (replylsn == InvalidXLogRecPtr) {
				elog(ERROR,"query pg_stat_replication replylsn failed");
				PQclear(pgres);
				return 1;
			} 
		} 
		//no slave,pushstandby no need wait
	} 
	else if (PQresultStatus(pgres) == PGRES_FATAL_ERROR) 
	{
		//master crash,pushstandby need replay to master crash point for private
		PQclear(pgres);
		return InvalidXLogRecPtr;
	}
	else 
	{
		PQfinish(pushconn);
		pushconn = NULL;
		PQclear(pgres);
		return 1;
	} 
	//elog(LOG,"appnamelsn: %x: replylsn %x",lsn,replylsn);
	if ((lsn !=InvalidXLogRecPtr && lsn < replylsn)||(replylsn == InvalidXLogRecPtr)) {
		replylsn = lsn;
	}
	PQclear(pgres);
	return replylsn;
}

// XLogRecPtr QueryReplyLsn(XLogRecPtr lsn) 
// {
// 	StringInfoData cmd;
// 	XLogRecPtr 		replylsn;
// 	PGresult   *pgres = NULL;
// 	char	   *appname;
// 	char       *state;
// 	char       *syncstate;
// 	char       *replyptr;
// 	replylsn = InvalidXLogRecPtr;
// 	if (pushconn == NULL) {
// 		if (ConnectPrimaryDB4ReplyLSN() == false) {
// 			return InvalidXLogRecPtr;
// 		} 
// 	} 

// 	initStringInfo(&cmd);
// 	appendStringInfoString(&cmd, "SELECT t.application_name, t.replay_lsn, t.state, t.sync_state FROM pg_catalog.pg_stat_replication t WHERE t.application_name <> \'");
// 	appendStringInfoString(&cmd, "pushstandby");
// 	appendStringInfoString(&cmd, "\' order by t.replay_lsn limit 1");

// 	pgres = PQexec(pushconn, cmd.data);
// 	if (PQresultStatus(pgres) == PGRES_TUPLES_OK && PQntuples(pgres) == 1) {
// 		appname = PQgetvalue(pgres, 0, 0);
// 		replyptr = PQgetvalue(pgres, 0, 1);
// 		bool flag;
// 		replylsn = pg_lsn_in_internal(replyptr,&flag); 
// 		//replylsn = atol(replyptr);
// 		state = PQgetvalue(pgres, 0, 2);
// 		syncstate = PQgetvalue(pgres, 0, 3);
// 	}
// 	else if (PQresultStatus(pgres) == PGRES_BAD_RESPONSE || 
// 		PQresultStatus(pgres) == PGRES_NONFATAL_ERROR ||
// 		PQresultStatus(pgres) == PGRES_FATAL_ERROR)
// 	{
// 		PQfinish(pushconn);
// 		pushconn = NULL;
// 		PQclear(pgres);
// 		return InvalidXLogRecPtr;
// 	}
// 	//elog(LOG,"appnamelsn: %x: replylsn %x",lsn,replylsn);
// 	if (lsn !=InvalidXLogRecPtr && lsn < replylsn||replylsn == InvalidXLogRecPtr) {
// 		replylsn = lsn;
// 	}
// 	PQclear(pgres);
// 	return replylsn;
// }
